#include <tiredis/Command.h>
#include <tiredis/Common.h>
#include <tiredis/UnboundedBuffer.h>
#include <tiredis/Executor.h>
#include <sstream>
#include <pingcap/kv/Scanner.h>
#include <tiredis/Statistics.h>

Error set(const std::vector<std::string>& params, UnboundedBuffer* reply)
{
    g_stat->setRC_ ++;
    pingcap::kv::Txn txn(Executor::tikv_cluster.get());
    txn.set(params[1], params[2]);
    txn.commit();
    FormatOK(reply);
    return Error_ok;
}

Error get(const std::vector<std::string>& params, UnboundedBuffer* reply)
{
    g_stat->getRC_ ++;
    pingcap::kv::Snapshot snap(Executor::tikv_cluster.get());
    std::string value = snap.Get(params[1]);
    if(value.back() == '0') {
        FormatNull(reply);
        return Error_ok;
    }
    value.pop_back();
    FormatBulk(value, reply);
    return Error_ok;
}

static Error ParseScanOption(const std::vector<std::string>& params, int start, long& count, const char*& pattern)
{
    // SCAN 0 [cursor] match * [pattern] count 60
    count = -1;
    pattern = nullptr;

    for (std::size_t i = start; i < params.size(); i += 2)
    {
        if (params[i].size() == 5)
        {
            if (strncasecmp(params[i].c_str(), "match", 5) == 0)
            {
                if (!pattern)
                {
                    pattern = params[i + 1].c_str();
                    continue;
                }
            }
            else if (strncasecmp(params[i].c_str(), "count", 5) == 0)
            {
                if(count == -1)
                {
                    std::stringstream sstream(params[i + 1]);
                    sstream >> count;
                    continue;
                }
            }
        }

        return Error_param;
    }

    return Error_ok;
}

Error scan(const std::vector<std::string>& params, UnboundedBuffer* reply)
{
    if (params.size() % 2 != 0)
    {
        ReplyError(Error_param, reply);
        return Error_param;
    }
    long cursor = 0;
    std::stringstream sstream(params[1]);
    sstream >> cursor;

    long count = -1;
    const char* pattern = nullptr;

    Error err = ParseScanOption(params, 2, count, pattern);
    if (err != Error_ok)
    {
        ReplyError(err, reply);
        return err;
    }

    if (count < 0) count = 5;
    std::vector<std::string> matchKeys;
    pingcap::kv::Snapshot snap(Executor::tikv_cluster.get());
    std::string prefix = "";

    if(pattern != nullptr)
    {
        prefix = std::string(pattern);
    }

    auto scanner = snap.Scan(prefix, pingcap::kv::prefixNext(prefix));
    long curNow = 0;
    while (cursor > 0) 
    {
        scanner.next();
        if(!scanner.valid) {
            break;
        }
        curNow ++;
        cursor --;
    }

    long step = 0;
    while(step < count && scanner.valid)
    {
        auto key = scanner.key();
        matchKeys.push_back(key);
        scanner.next();
        curNow ++;
        step ++;
    }
    if (step < count) {
        curNow = 0;
    }

    PreFormatMultiBulk(2, reply);

    char buf[32];
    auto len = snprintf(buf, sizeof buf - 1, "%lu", curNow);
    FormatBulk(buf, len, reply); // cursor

    PreFormatMultiBulk(matchKeys.size(), reply);
    // keys
    for (const auto& s : matchKeys)
    {
        FormatBulk(s, reply);
    }

    g_stat->scanRC_ ++;
    return Error_ok;
}
